package listen

import (
	"gitee.com/zaper/kafka-service/service/consumer"
	"gitee.com/zaper/kafka-service/service/consumer/consumer_model"
	logger "gitee.com/zaper/private-logger"
	"gitee.com/zaper/service-config-client/os_env"
	"os"
	"time"
)

func ConfigListen(topicName string) {
	cSum := consumer.NewConsumerAgent()
	defer func() {
		err := cSum.Close()
		if err != nil {
			logger.Errorf("close kafka consumer fail, err is %s", err)
		}
	}()
	for {
		LoopCheckKafkaTopicOnline(topicName, 5)
		_ = cSum.ConsumerSub(&consumer_model.ConsumerModel{
			Topic:   topicName,
			GroupID: os.Getenv(os_env.ServiceIdKey),
		})
		consumerChan, err := cSum.ConsumerMess()
		if err != nil {
			logger.Infof("kafka ConsumerMess fail, err %s", err)
			time.Sleep(30 * time.Second)
		}
		listenMgr := NewLisManager(consumerChan)
		go listenMgr.HandlerKafka()
		listenMgr.HandlerMessSchedule()
		// 监听到异常时自行修复
		<-listenMgr.ListenErrChan
		time.Sleep(10 * time.Second)
	}
}
